/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.geode.spark.connector

import org.apache.geode.spark.connector.internal.rdd.{GeodeOuterJoinRDD, GeodeJoinRDD, GeodePairRDDWriter}
import org.apache.spark.Logging
import org.apache.spark.api.java.function.Function
import org.apache.spark.rdd.RDD

/**
 * Extra gemFire functions on RDDs of (key, value) pairs through an implicit conversion.
 * Import `org.apache.geode.spark.connector._` at the top of your program to
 * use these functions.
 */
class GeodePairRDDFunctions[K, V](val rdd: RDD[(K, V)]) extends Serializable with Logging {

  /**
   * Save the RDD of pairs to Geode key-value store without any conversion
   * @param regionPath the full path of region that the RDD is stored
   * @param connConf the GeodeConnectionConf object that provides connection to Geode cluster
   * @param opConf the optional parameters for this operation
   */
  def saveToGeode(
      regionPath: String, 
      connConf: GeodeConnectionConf = defaultConnectionConf, 
      opConf: Map[String, String] = Map.empty): Unit = {    
    connConf.getConnection.validateRegion[K, V](regionPath)
    if (log.isDebugEnabled)
      logDebug(s"""Save RDD id=${rdd.id} to region $regionPath, partitions:\n  ${getRddPartitionsInfo(rdd)}""")
    else
      logInfo(s"""Save RDD id=${rdd.id} to region $regionPath""")
    val writer = new GeodePairRDDWriter[K, V](regionPath, connConf, opConf)
    rdd.sparkContext.runJob(rdd, writer.write _)
  }

  /**
   * Return an RDD containing all pairs of elements with matching keys in `this`
   * RDD and the Geode `Region[K, V2]`. Each pair of elements will be returned
   * as a ((k, v), v2) tuple, where (k, v) is in `this` RDD and (k, v2) is in the
   * Geode region.
   *
   *@param regionPath the region path of the Geode region
   * @param connConf the GeodeConnectionConf object that provides connection to Geode cluster
   * @tparam K2 the key type of the Geode region
   * @tparam V2 the value type of the Geode region
   * @return RDD[T, V]
   */
  def joinGeodeRegion[K2 <: K, V2](
    regionPath: String, connConf: GeodeConnectionConf = defaultConnectionConf): GeodeJoinRDD[(K, V), K, V2] = {
    new GeodeJoinRDD[(K, V), K, V2](rdd, null, regionPath, connConf)
  }

  /**
   * Return an RDD containing all pairs of elements with matching keys in `this` RDD
   * and the Geode `Region[K2, V2]`. The join key from RDD element is generated by
   * `func(K, V) => K2`, and the key from the Geode region is jus the key of the
   * key/value pair.
   *
   * Each pair of elements of result RDD will be returned as a ((k, v), v2) tuple,
   * where (k, v) is in `this` RDD and (k2, v2) is in the Geode region.
   *
   * @param regionPath the region path of the Geode region
   * @param func the function that generates region key from RDD element (K, V)
   * @param connConf the GeodeConnectionConf object that provides connection to Geode cluster
   * @tparam K2 the key type of the Geode region
   * @tparam V2 the value type of the Geode region
   * @return RDD[(K, V), V2]
   */
  def joinGeodeRegion[K2, V2](
    regionPath: String, func: ((K, V)) => K2, connConf: GeodeConnectionConf = defaultConnectionConf): GeodeJoinRDD[(K, V), K2, V2] =
    new GeodeJoinRDD[(K, V), K2, V2](rdd, func, regionPath, connConf)

  /** This version of joinGeodeRegion(...) is just for Java API. */
  private[connector] def joinGeodeRegion[K2, V2](
    regionPath: String, func: Function[(K, V), K2], connConf: GeodeConnectionConf): GeodeJoinRDD[(K, V), K2, V2] = {
    new GeodeJoinRDD[(K, V), K2, V2](rdd, func.call, regionPath, connConf)
  }

  /**
   * Perform a left outer join of `this` RDD and the Geode `Region[K, V2]`.
   * For each element (k, v) in `this` RDD, the resulting RDD will either contain
   * all pairs ((k, v), Some(v2)) for v2 in the Geode region, or the pair
   * ((k, v), None)) if no element in the Geode region have key k.
   *
   * @param regionPath the region path of the Geode region
   * @param connConf the GeodeConnectionConf object that provides connection to Geode cluster
   * @tparam K2 the key type of the Geode region
   * @tparam V2 the value type of the Geode region
   * @return RDD[ (K, V), Option[V] ]
   */
  def outerJoinGeodeRegion[K2 <: K, V2](
    regionPath: String, connConf: GeodeConnectionConf = defaultConnectionConf): GeodeOuterJoinRDD[(K, V), K, V2] = {
    new GeodeOuterJoinRDD[(K, V), K, V2](rdd, null, regionPath, connConf)
  }

  /**
   * Perform a left outer join of `this` RDD and the Geode `Region[K2, V2]`.
   * The join key from RDD element is generated by `func(K, V) => K2`, and the
   * key from region is jus the key of the key/value pair.
   *
   * For each element (k, v) in `this` RDD, the resulting RDD will either contain
   * all pairs ((k, v), Some(v2)) for v2 in the Geode region, or the pair
   * ((k, v), None)) if no element in the Geode region have key `func(k, v)`.
   *
   *@param regionPath the region path of the Geode region
   * @param func the function that generates region key from RDD element (K, V)
   * @param connConf the GeodeConnectionConf object that provides connection to Geode cluster
   * @tparam K2 the key type of the Geode region
   * @tparam V2 the value type of the Geode region
   * @return RDD[ (K, V), Option[V] ]
   */
  def outerJoinGeodeRegion[K2, V2](
    regionPath: String, func: ((K, V)) => K2, connConf: GeodeConnectionConf = defaultConnectionConf): GeodeOuterJoinRDD[(K, V), K2, V2] = {
    new GeodeOuterJoinRDD[(K, V), K2, V2](rdd, func, regionPath, connConf)
  }

  /** This version of outerJoinGeodeRegion(...) is just for Java API. */
  private[connector] def outerJoinGeodeRegion[K2, V2](
    regionPath: String, func: Function[(K, V), K2], connConf: GeodeConnectionConf): GeodeOuterJoinRDD[(K, V), K2, V2] = {
    new GeodeOuterJoinRDD[(K, V), K2, V2](rdd, func.call, regionPath, connConf)
  }

  private[connector] def defaultConnectionConf: GeodeConnectionConf =
    GeodeConnectionConf(rdd.sparkContext.getConf)

}
